package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.termalRunaway.*;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.constant.TermalProcessResult;
import cn.gwm.flink.streaming.constant.TermalResultEnum;
import cn.gwm.flink.streaming.constant.VehiclePowerType;
import cn.gwm.flink.streaming.task.ThermalWarningEvDwmTaskTest;
import cn.gwm.utils.StreamUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Dict;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 热失控返回结果组装
 * @author GW00283474
 */
public class TermalWarningResultEvProcessFunction extends KeyedProcessFunction<String, Dict, String> {
    private static final Logger logger = LoggerFactory.getLogger(TermalWarningResultEvProcessFunction.class);
    /**
     * 热失控报警规则匹配数量
     */
    private int alarmRulePatternNum = 2;
    /**
     * 热失控预警规则匹配数量
     */
    private int forewarnRulePatternNum = 1;

    /**
     * 经过云端检测出来的热失控事件源信号集合
     */
    private OutputTag<JSONObject> outputTag;


    private VehiclePowerType vehiclePowerType;

    private MapState<String, WaringTermalDwmWithStragegyEv> singlesState;

    public TermalWarningResultEvProcessFunction(int alarmRulePatternNum, int forewarnRulePatternNum, OutputTag<JSONObject> outputTag,VehiclePowerType vehiclePowerType) {
        this.alarmRulePatternNum = alarmRulePatternNum;
        this.forewarnRulePatternNum = forewarnRulePatternNum;
        this.outputTag=outputTag;
        this.vehiclePowerType=vehiclePowerType;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        StateTtlConfig stateTtlConfig = StateTtlConfig
                // 状态有效时间    1天过期
                .newBuilder(Time.days(1))
                // 设置状态的更新类型
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 已过期还未被清理掉的状态数据不返回给用户
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                // 过期对象的清理策略 全量清理
                .cleanupFullSnapshot()
                .build();
        MapStateDescriptor singlesDesc = new MapStateDescriptor("singlesDesc"+this.vehiclePowerType.name(), String.class, WaringTermalDwmWithStragegyEv.class);
        singlesDesc.enableTimeToLive(stateTtlConfig);
        singlesState = getRuntimeContext().getMapState(singlesDesc);
    }

    @Override
    public void processElement(Dict dict, KeyedProcessFunction<String, Dict, String>.Context ctx, Collector<String> out) throws Exception {
        try {
            //返回对象
            TermalRunawayRes res= new TermalRunawayRes();
            res.setRegion("未知");
            //云端检测
            WaringTermalWithStragegyAndFlagEv record=(WaringTermalWithStragegyAndFlagEv) dict.get(TermalProcessResult.CHECK_RECORD);
            Dict tempMaxRulePatternDict=(Dict) dict.get(TermalProcessResult.DICT_TEMP);
            Dict voltRateRulePatternDict=(Dict) dict.get(TermalProcessResult.DICT_VOLTCHANAGE);
            Dict tempRateRulePatternDict=(Dict) dict.get(TermalProcessResult.DICT_TEMPRISE);
            Dict lowVoltRulePatternDict=(Dict) dict.get(TermalProcessResult.DICT_VOLTAGE);
            Dict tempMaxDeviationRulePatternDict=(Dict)dict.get(TermalProcessResult.DICT_TEMPDIFF);
            //车端上传的热失控事件
            Dict termalVehicleUploadRulePatternDict=(Dict)dict.get(TermalProcessResult.DICT_TERMAL_UPLOAD);

            String processTime=DateUtil.format(new Date(ctx.timerService().currentProcessingTime()),"yyyy-MM-dd HH:mm:ss");

            //模组的最高温度大于65，并持续30s及以上
            boolean tempMaxRulePattern = record.getTemperatureWarn();
            //单体电压速率下降超过0.83%/s,且不低于30s下降25%
            boolean voltRateRulePattern = record.getVoltageVariationWarn();
            //模组的温度温升速率超过1℃/3s，且不低于30s上升10度
            boolean tempRateRulePattern = record.getTemperatureRiseWarn();
            //单电压<1.5V 并且持续≥30s
            boolean lowVoltRulePattern = record.getVoltageWarn();
            //电池组模组温差大于等于25度,持续30s及以上
            boolean tempMaxDeviationRulePattern = record.getTemperatureDifferenceWarn();
            //车端上传热失控事件
            boolean termalRunawayWarnRulePattern = record.getTermalRunawayWarn();

            int rulePattern = (int) ListUtil.of(tempMaxRulePattern, voltRateRulePattern, tempRateRulePattern, lowVoltRulePattern, tempMaxDeviationRulePattern).stream().filter(bool -> bool.equals(true)).count();
            //云端检测热失控
            boolean isBmsAlert = rulePattern >= alarmRulePatternNum;
            boolean isWaring= rulePattern == forewarnRulePatternNum;
            boolean isNormal = rulePattern==0;
            //最终热失控事件标识
            boolean isAlert = BooleanUtil.or(isBmsAlert, termalRunawayWarnRulePattern);
            //热失控预警标识
            isWaring = BooleanUtil.and(isWaring, !termalRunawayWarnRulePattern);
            //正常数据标识
            isNormal =BooleanUtil.and(isNormal,!termalRunawayWarnRulePattern);
            res.setModel_name(record.getVehicletype());
            res.setModel_code(record.getModel_code());
            //检测日期
            res.setDateTime(record.getIngestionDate());
            res.setDeviceId(record.getDeviceId());
            res.setLatitude(record.getLatitude());
            res.setLongitude(record.getLongitude());
            res.setTestTime(record.getIngestionTime());
            //正常数据
            if(isNormal){
                //1:云端检测  2:车端上传
                res.getDataSource().add("1");
                res.setResult(TermalResultEnum.NORMAL.getKey());
                res.setBmsTrdAbnormals(new ArrayList<BmsTrdAbnormal>(0));
                res.setBmsTrdSignals(new ArrayList<BmsTrdSignal>(0));
            }
            //热失控预警
            if(isWaring){
                //1:云端检测  2:车端上传
                res.getDataSource().add("1");
                res.setResult(TermalResultEnum.WARNING.getKey());
                List<BmsTrdAbnormal> bmsTrdAbnormals=new ArrayList<>(0);
                writeBmsRulePatterns(tempMaxRulePattern,voltRateRulePattern,tempRateRulePattern,lowVoltRulePattern,tempMaxDeviationRulePattern
                        ,tempMaxRulePatternDict,voltRateRulePatternDict,tempRateRulePatternDict,lowVoltRulePatternDict,tempMaxDeviationRulePatternDict
                        ,res,record,processTime
                        ,bmsTrdAbnormals);
                res.setBmsTrdSignals(new ArrayList<BmsTrdSignal>(0));
            }
            //最终热失控事件
            if(isAlert){
                res.setResult(TermalResultEnum.EVENT.getKey());
                long recentlyTid=Long.MIN_VALUE;
                List<BmsTrdAbnormal> bmsTrdAbnormals=new ArrayList<>(0);
                //云端检测热失控事件
                if(BooleanUtil.isTrue(isBmsAlert)){
                    //1:云端检测  2:车端上传
                    res.getDataSource().add("1");
                    writeBmsRulePatterns(tempMaxRulePattern,voltRateRulePattern,tempRateRulePattern,lowVoltRulePattern,tempMaxDeviationRulePattern
                            ,tempMaxRulePatternDict,voltRateRulePatternDict,tempRateRulePatternDict,lowVoltRulePatternDict,tempMaxDeviationRulePatternDict
                            ,res,record,processTime
                            ,bmsTrdAbnormals);

                    if(tempMaxRulePattern){
                        recentlyTid = NumberUtil.max((long)tempMaxRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                        List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)tempMaxRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                        checkSinglesState(list,singlesState);
                    }
                    if(voltRateRulePattern){
                        recentlyTid = NumberUtil.max((long)voltRateRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                        List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)voltRateRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                        checkSinglesState(list,singlesState);
                    }
                    if(tempRateRulePattern){
                        recentlyTid = NumberUtil.max((long)tempRateRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                        List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)tempRateRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                        checkSinglesState(list,singlesState);
                    }
                    if(lowVoltRulePattern){
                        recentlyTid = NumberUtil.max((long)lowVoltRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                        List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)lowVoltRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                        checkSinglesState(list,singlesState);
                    }
                    if(tempMaxDeviationRulePattern){
                        recentlyTid = NumberUtil.max((long)tempMaxDeviationRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                        List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)tempMaxDeviationRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                        checkSinglesState(list,singlesState);
                    }


                }
                //车端上传热失控事件
                if(BooleanUtil.isTrue(termalRunawayWarnRulePattern)){
                    //1：云端检测  2：车端上传
                    res.getDataSource().add("2");
                    recentlyTid = NumberUtil.max((long)termalVehicleUploadRulePatternDict.get(TermalProcessResult.RECENTLY_TID),recentlyTid);
                    writeVehicleRulePatterns(termalRunawayWarnRulePattern,termalVehicleUploadRulePatternDict,res,record,processTime,bmsTrdAbnormals);
                    List<WaringTermalDwmWithStragegyEv> list=(List<WaringTermalDwmWithStragegyEv>)termalVehicleUploadRulePatternDict.get(TermalProcessResult.CHECK_RECORDS);
                    checkSinglesState(list,singlesState);
                }

                List<WaringTermalDwmWithStragegyEv> singles = StreamUtil.convertIterable(singlesState.values());
                List<WaringTermalDwmWithStragegyEv> collect = singles.stream().filter(s -> !s.checkFieldNullValue()).collect(Collectors.toList());
                collect.stream().forEach(s-> ctx.output(outputTag,s.getPropertiesMap()));
                res.setItem_time(recentlyTid);
                singlesState.clear();
            }
            out.collect(JSONUtil.toJsonStr(res));
        }catch (Exception e){
            e.printStackTrace();
            logger.error("热失控检测函数TermalWarningResultEvProcessFunction处理异常message:{},case:{},stackTrace:{}",e.getMessage(),e.getCause(),e.getStackTrace());
        }
    }

    //云端检测的热失控规则数据返回
    private static void writeBmsRulePatterns(Boolean tempMaxRulePattern, Boolean voltRateRulePattern, Boolean tempRateRulePattern
            , Boolean lowVoltRulePattern, Boolean tempMaxDeviationRulePattern, Dict tempMaxRulePatternDict
            , Dict voltRateRulePatternDict, Dict tempRateRulePatternDict
            , Dict lowVoltRulePatternDict, Dict tempMaxDeviationRulePatternDict
            , TermalRunawayRes res, WaringTermalWithStragegyAndFlagEv current, String processTime
            , List<BmsTrdAbnormal> bmsTrdAbnormals) throws Exception{
        String exceptionType="";
        if(tempMaxRulePattern){
            //最大温度报警
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setMaxTemperature(tempMaxRulePatternDict.get(TermalProcessResult.MAX_TEMPERATURE,0).toString());
            trdAbnormal.setMinTemperature(tempMaxRulePatternDict.get(TermalProcessResult.MIN_TEMPERATURE,0).toString());
            trdAbnormal.setTestItem("温度");
            trdAbnormal.setThreshold(tempMaxRulePatternDict.get(TermalProcessResult.THRESHOLD_MAX_TEMP,new BigDecimal(0)).toString());
            trdAbnormal.setWarningValue(tempMaxRulePatternDict.get(TermalProcessResult.MAX_TEMPERATURE,0).toString());
            trdAbnormal.setWarningElectric(Objects.nonNull(current.getBMS_SOC())?current.getBMS_SOC().toString():null);
            trdAbnormal.setWarningContent("温度");
            trdAbnormal.setWarningTime(processTime);
            bmsTrdAbnormals.add(trdAbnormal);
            exceptionType+=tempMaxRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString();
        }
        if(voltRateRulePattern){
            //电压变化速率报警
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setMaxVoltage(voltRateRulePatternDict.get(TermalProcessResult.MAX_VOLTAGE,new BigDecimal(0)).toString());
            trdAbnormal.setMinVoltage(voltRateRulePatternDict.get(TermalProcessResult.MIN_VOLTAGE,new BigDecimal(0)).toString());
            trdAbnormal.setTestItem("电压变化");
            trdAbnormal.setThreshold(voltRateRulePatternDict.get(TermalProcessResult.THRESHOLD_VOLTAGE_CHANGERISE,new BigDecimal(0)).toString());
            trdAbnormal.setWarningValue(voltRateRulePatternDict.get(TermalProcessResult.MAX_VOLTAGE_CHANGERISE,new BigDecimal(0)).toString());
            trdAbnormal.setWarningElectric(Objects.nonNull(current.getBMS_SOC())?current.getBMS_SOC().toString():null);
            trdAbnormal.setWarningContent("电压变化");
            trdAbnormal.setWarningTime(processTime);
            trdAbnormal.setMaxVoltageChange(voltRateRulePatternDict.get(TermalProcessResult.MAX_VOLTAGE_CHANGERISE,0.00).toString());
            bmsTrdAbnormals.add(trdAbnormal);
            if(StringUtils.isNotEmpty(exceptionType)){
                exceptionType+=(","+voltRateRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString());
            }
        }
        if(tempRateRulePattern){
            //模组升温速率报警
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setMaxTemperature(tempRateRulePatternDict.get(TermalProcessResult.MAX_TEMPERATURE,0).toString());
            trdAbnormal.setMinTemperature(tempRateRulePatternDict.get(TermalProcessResult.MIN_TEMPERATURE,0).toString());
            trdAbnormal.setTestItem("温升");
            trdAbnormal.setThreshold(tempRateRulePatternDict.get(TermalProcessResult.THRESHOLD_TEMPRISE,new BigDecimal(0)).toString());
            trdAbnormal.setWarningValue(tempRateRulePatternDict.get(TermalProcessResult.MAX_TEMPRISE, 0.00).toString());
            trdAbnormal.setWarningElectric(Objects.nonNull(current.getBMS_SOC())?current.getBMS_SOC().toString():null);
            trdAbnormal.setWarningContent("温升");
            trdAbnormal.setWarningTime(processTime);
            trdAbnormal.setMaxTemperatureRise(tempRateRulePatternDict.get(TermalProcessResult.MAX_TEMPRISE,0.00).toString());
            bmsTrdAbnormals.add(trdAbnormal);
            if(StringUtils.isNotEmpty(exceptionType)){
                exceptionType+=(","+tempRateRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString());
            }
        }
        if(lowVoltRulePattern){
            //最小电压报警
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setMaxVoltage(lowVoltRulePatternDict.get(TermalProcessResult.MAX_VOLTAGE, new BigDecimal(0)).toString());
            trdAbnormal.setMinVoltage(lowVoltRulePatternDict.get(TermalProcessResult.MIN_VOLTAGE,new BigDecimal(0)).toString());
            trdAbnormal.setTestItem("电压");
            trdAbnormal.setThreshold(lowVoltRulePatternDict.get(TermalProcessResult.THRESHOLD_MIN_VOLTAGE,new BigDecimal(0)).toString());
            trdAbnormal.setWarningValue(lowVoltRulePatternDict.get(TermalProcessResult.MIN_VOLTAGE,new BigDecimal(0)).toString());
            trdAbnormal.setWarningElectric(Objects.nonNull(current.getBMS_SOC())?current.getBMS_SOC().toString():null);
            trdAbnormal.setWarningContent("电压");
            trdAbnormal.setWarningTime(processTime);
            bmsTrdAbnormals.add(trdAbnormal);
            if(StringUtils.isNotEmpty(exceptionType)){
                exceptionType+=(","+lowVoltRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString());
            }
        }
        if(tempMaxDeviationRulePattern){
            //温差报警
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setMaxTemperature(tempMaxDeviationRulePatternDict.get(TermalProcessResult.MAX_TEMPERATURE,0).toString());
            trdAbnormal.setMinTemperature(tempMaxDeviationRulePatternDict.get(TermalProcessResult.MIN_TEMPERATURE,0).toString());
            trdAbnormal.setTestItem("温差");
            trdAbnormal.setThreshold(tempMaxDeviationRulePatternDict.get(TermalProcessResult.THRESHOLD_TEMPDIFF,new BigDecimal(0)).toString());
            trdAbnormal.setWarningValue(String.valueOf(Integer.parseInt(tempMaxDeviationRulePatternDict.get(TermalProcessResult.MAX_TEMPERATURE,0).toString())-Integer.parseInt(tempMaxDeviationRulePatternDict.get(TermalProcessResult.MIN_TEMPERATURE,0).toString())));
            trdAbnormal.setWarningElectric(Objects.nonNull(current.getBMS_SOC())?current.getBMS_SOC().toString():null);
            trdAbnormal.setWarningContent("温差");
            trdAbnormal.setWarningTime(processTime);
            bmsTrdAbnormals.add(trdAbnormal);
            if(StringUtils.isNotEmpty(exceptionType)){
                exceptionType+=(","+tempMaxDeviationRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString());
            }
        }
        res.setBmsTrdAbnormals(bmsTrdAbnormals);
        if(StringUtils.isNotEmpty(exceptionType)){
            current.setExceptionType(exceptionType);
        }
    }

    //车端上传的热失控事件数据返回
    private static void writeVehicleRulePatterns(Boolean termalRunawayWarnRulePattern,Dict termalVehicleUploadRulePatternDict
            , TermalRunawayRes res, WaringTermalWithStragegyAndFlagEv current, String processTime
            , List<BmsTrdAbnormal> bmsTrdAbnormals) throws Exception{
        String exceptionType=current.getExceptionType();
        if(termalRunawayWarnRulePattern){
            BmsTrdAbnormal trdAbnormal= new BmsTrdAbnormal();
            trdAbnormal.setTestItem("车端上传热失控事件");
            trdAbnormal.setWarningContent("车端上传热失控事件");
            trdAbnormal.setWarningValue(String.valueOf(termalVehicleUploadRulePatternDict.get(TermalProcessResult.TERMAL_EVENT_CODE, String.valueOf(DefaultConstant.BMS_BattHeatRunaway.intValue()))));
            trdAbnormal.setWarningTime(processTime);
            bmsTrdAbnormals.add(trdAbnormal);
            if(StringUtils.isNotEmpty(exceptionType)){
                exceptionType+=(","+termalVehicleUploadRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString());
            }else{
                exceptionType+=termalVehicleUploadRulePatternDict.get(TermalProcessResult.STRATEGY_DESC).toString();
            }
        }
        res.setBmsTrdAbnormals(bmsTrdAbnormals);
        if(StringUtils.isNotEmpty(exceptionType)){
            current.setExceptionType(exceptionType);
        }

    }



    private void checkSinglesState(List<WaringTermalDwmWithStragegyEv> list, MapState<String, WaringTermalDwmWithStragegyEv> state) throws Exception{
        for(WaringTermalDwmWithStragegyEv element:list){
            String key=element.getDeviceId()+"-"+element.getItem_time();
            List<String> keys = StreamUtil.convertIterable(state.keys());
            if(!keys.contains(key)){
                state.put(key,element);
            }
        }
    }
}
